package omega

import (
	"context"
	"fmt"
	"reflect"

	"gitee.com/zawei/yggdrasil/pkg/lib/omega/proto"
	"gitee.com/zawei/yggdrasil/pkg/log"
	"github.com/pkg/errors"
)

func getOmegaCtx(ctx context.Context) (*OmegaContext, error) {
	omegaCtx, ok := ctx.Value(omegaCtxKey{}).(*OmegaContext)
	if !ok {
		return omegaCtx, errors.New("not found omega context")
	}
	return omegaCtx, nil
}

func SagaStartBeforeInject(ctx context.Context, timeout int, targetMethod string) (context.Context, error) {
	uuid := getUuid()
	ctx = context.WithValue(ctx, omegaCtxKey{}, &OmegaContext{
		GlobalTxId: uuid,
		LocalTxId:  uuid,
	})
	_, err := transportInstance.txOnTxEvent(ctx, &proto.GrpcTxEvent{
		Type:               SagaStartedEvent,
		Timeout:            int32(timeout),
		CompensationMethod: targetMethod,
	})
	if err != nil {
		_, _ = transportInstance.sendSagaAbortedEvent(ctx, targetMethod, err)
		return nil, err
	}

	return ctx, nil
}

func SagaEndAfterInject(ctx context.Context, targetMethod string, err error, autoClose bool) error {
	if autoClose {
		return SendSagaEndEvent(ctx, targetMethod, err)
	}
	return nil
}

func SendSagaEndEvent(ctx context.Context, method string, err error) error {
	omegaCtx, ok := ctx.Value(omegaCtxKey{}).(*OmegaContext)
	if !ok && err == nil {
		err = errors.New("not fount omega context")
	}
	if err != nil {
		if !config.AkkaEnabled {
			_, _ = transportInstance.sendTxAbortedEvent(ctx, method, err)
		} else {
			_, _ = transportInstance.sendSagaAbortedEvent(ctx, method, err)
		}
		return err
	}
	aborted, err := transportInstance.txOnTxEvent(ctx, &proto.GrpcTxEvent{
		Type: SagaEndedEvent,
	})
	if err != nil {
		log.Errorf("fault to end saga, error: %s", err.Error())
		if !config.AkkaEnabled {
			_, _ = transportInstance.sendTxAbortedEvent(ctx, method, err)
		} else {
			_, _ = transportInstance.sendSagaAbortedEvent(ctx, method, err)
		}
	}

	if aborted {
		return errors.New(fmt.Sprintf("transaction %s is aborted", omegaCtx.GlobalTxId))
	}
	log.Debugf("saga method[%s] finish", method)
	return nil
}

func TxStartBeforeInject(ctx context.Context, targetMethod, compensateMethod string, compensateArgs []reflect.Value, timeout int) (context.Context, error) {
	omegaCtx, err := getOmegaCtx(ctx)
	if err != nil {
		return nil, err
	}
	parentID := omegaCtx.LocalTxId
	localId := getUuid()
	defer func() {
		if err := recover(); err != nil {
			omegaCtx.LocalTxId = parentID
		}
	}()
	omegaCtx.ParentTxId = omegaCtx.LocalTxId
	omegaCtx.LocalTxId = localId

	payload, err := config.encode.Marshal(compensateArgs)
	if err != nil {
		omegaCtx.ParentTxId = parentID
		return nil, err
	}
	aborted, err := transportInstance.txOnTxEvent(ctx, &proto.GrpcTxEvent{
		Type:               TxStartedEvent,
		Timeout:            int32(timeout),
		CompensationMethod: compensateMethod,
		Payloads:           payload,
	})
	if err != nil {
		omegaCtx.ParentTxId = parentID
		return nil, err
	}
	if aborted {
		omegaCtx.ParentTxId = parentID
		return nil, errors.New(fmt.Sprintf("Abort sub transaction %s because global transaction %s has already aborted.", omegaCtx.LocalTxId, omegaCtx.GlobalTxId))
	}

	return ctx, nil
}

func TxEndAfterInject(ctx context.Context, targetMethod, compensateMethod string, err error) error {
	omegaCtx, err1 := getOmegaCtx(ctx)
	if err1 != nil {
		return err1
	}
	if err != nil {
		_, _ = transportInstance.sendTxAbortedEvent(ctx, targetMethod, err)
		omegaCtx.LocalTxId = omegaCtx.ParentTxId
		return err
	}
	aborted, err := transportInstance.txOnTxEvent(ctx, &proto.GrpcTxEvent{
		Type:               TxEndedEvent,
		CompensationMethod: compensateMethod,
	})
	if err != nil {
		omegaCtx.LocalTxId = omegaCtx.ParentTxId
		_, _ = transportInstance.sendTxAbortedEvent(ctx, targetMethod, err)
		return err
	}
	if aborted {
		omegaCtx.LocalTxId = omegaCtx.ParentTxId
		return errors.New(fmt.Sprintf("transaction %s is aborted", omegaCtx.LocalTxId))
	}
	return nil
}

//-------------------------------------------------------

func TccStartBeforeInject(ctx context.Context) (context.Context, error) {
	uuid := getUuid()
	ctx = context.WithValue(ctx, omegaCtxKey{}, &OmegaContext{
		GlobalTxId: uuid,
		LocalTxId:  uuid,
	})
	aborted, err := transportInstance.OnTccTransactionStarted(ctx)
	if err != nil {
		_, _ = transportInstance.OnTccTransactionEnded(ctx, TransactionFailed)
		return nil, err
	}
	if aborted {
		return nil, errors.New(fmt.Sprintf("transaction %s is aborted", uuid))
	}
	return ctx, nil
}

func TccEndAfterInject(ctx context.Context, targetMethod string, err error, autoClose bool) error {
	if autoClose {
		return SendTccEndEvent(ctx, targetMethod, err)
	}
	return nil
}

func SendTccEndEvent(ctx context.Context, method string, err error) error {
	omegaCtx, ok := ctx.Value(omegaCtxKey{}).(*OmegaContext)
	if !ok && err == nil {
		err = errors.New("not fount omega context")
	}
	if err != nil {
		_, _ = transportInstance.OnTccTransactionEnded(ctx, TransactionFailed)
		return err
	}
	aborted, err := transportInstance.OnTccTransactionEnded(ctx, TransactionSucceed)
	if err != nil {
		log.Errorf("fault to end tcc transaction, error: %s", err.Error())
		_, _ = transportInstance.OnTccTransactionEnded(ctx, TransactionFailed)
	}
	if aborted {
		return errors.New(fmt.Sprintf("transaction %s is aborted", omegaCtx.GlobalTxId))
	}
	log.Debugf("tcc transaction method[%s] finish", method)
	return nil
}

func TccTryStartBefore(ctx context.Context, confirmMethod, cancelMethod string) (context.Context, error) {
	omegaCtx, err := getOmegaCtx(ctx)
	if err != nil {
		return nil, err
	}
	parentID := omegaCtx.LocalTxId
	localId := getUuid()
	defer func() {
		if err := recover(); err != nil {
			omegaCtx.LocalTxId = parentID
		}
	}()
	omegaCtx.ParentTxId = omegaCtx.LocalTxId
	omegaCtx.LocalTxId = localId
	aborted, err := transportInstance.OnParticipationStarted(ctx, confirmMethod, cancelMethod)
	if err != nil {
		omegaCtx.ParentTxId = parentID
		return nil, err
	}
	if aborted {
		omegaCtx.ParentTxId = parentID
		return nil, errors.New(fmt.Sprintf("Abort sub transaction %s because global transaction %s has already aborted.", omegaCtx.LocalTxId, omegaCtx.GlobalTxId))
	}

	return ctx, nil
}

func TccTryEndAfterInject(ctx context.Context, confirmMethod, cancelMethod string, err error) error {
	omegaCtx, err1 := getOmegaCtx(ctx)
	if err1 != nil {
		return err1
	}
	if err != nil {
		_, _ = transportInstance.OnParticipationEnded(ctx, confirmMethod, cancelMethod, TransactionFailed)
		omegaCtx.LocalTxId = omegaCtx.ParentTxId
		return err
	}
	aborted, err := transportInstance.OnParticipationEnded(ctx, confirmMethod, cancelMethod, TransactionSucceed)
	if err != nil {
		omegaCtx.LocalTxId = omegaCtx.ParentTxId
		return err
	}
	if aborted {
		omegaCtx.LocalTxId = omegaCtx.ParentTxId
		return errors.New(fmt.Sprintf("transaction %s is aborted", omegaCtx.LocalTxId))
	}
	return nil
}
